package nats

import (
	"context"
	"errors"
	"micro-rulego/internal/broker"
	"strings"
	"sync"
	"time"
	"unicode"

	"github.com/nats-io/nats.go"
)

type jsPublication struct {
	t   string
	err error
	m   []byte
	ctx context.Context
	msg *nats.Msg
}

func (p *jsPublication) Topic() string            { return p.t }
func (p *jsPublication) Message() []byte          { return p.m }
func (p *jsPublication) Context() context.Context { return p.ctx }
func (p *jsPublication) Error() error             { return p.err }
func (p *jsPublication) Ack() error               { return p.msg.Ack() }

type jsSubscriber struct {
	sub  *nats.Subscription
	opts broker.SubscribeOptions
}

func (s *jsSubscriber) Options() broker.SubscribeOptions { return s.opts }
func (s *jsSubscriber) Topic() string                    { return s.sub.Subject }
func (s *jsSubscriber) Unsubscribe() error               { return s.sub.Unsubscribe() }

type jetStreamBroker struct {
	sync.Once
	sync.RWMutex

	connected bool
	conn      *nats.Conn
	js        nats.JetStreamContext
	opts      broker.Options
	addrs     []string
}

func (j *jetStreamBroker) Init(opts ...broker.Option) error {
	for _, o := range opts {
		o(&j.opts)
	}
	j.addrs = j.setAddrs(j.opts.Addrs)
	return nil
}

func (j *jetStreamBroker) Options() broker.Options { return j.opts }
func (j *jetStreamBroker) String() string          { return "nats-jetstream" }

func (j *jetStreamBroker) Address() string {
	if j.conn != nil && j.conn.IsConnected() {
		return j.conn.ConnectedUrl()
	}
	if len(j.addrs) > 0 {
		return j.addrs[0]
	}
	return ""
}

func (j *jetStreamBroker) Connect() error {
	j.Lock()
	defer j.Unlock()

	if j.connected {
		return nil
	}

	opts := nats.GetDefaultOptions()
	opts.Servers = j.addrs
	opts.Secure = j.opts.Secure
	opts.TLSConfig = j.opts.TLSConfig

	conn, err := opts.Connect()
	if err != nil {
		return err
	}

	js, err := conn.JetStream()
	if err != nil {
		conn.Close()
		return err
	}

	j.conn = conn
	j.js = js
	j.connected = true
	return nil
}

func (j *jetStreamBroker) Disconnect() error {
	j.Lock()
	defer j.Unlock()

	if j.conn != nil {
		j.conn.Close()
	}
	j.connected = false
	return nil
}

func createStreamName(topic string) string {
	// 替换非法字符，只保留字母、数字和下划线
	streamName := strings.Map(func(r rune) rune {
		if unicode.IsLetter(r) || unicode.IsNumber(r) || r == '_' || r == '-' {
			return r
		}
		return '_'
	}, topic)

	// 确保名称以字母开头
	if !unicode.IsLetter(rune(streamName[0])) {
		streamName = "stream_" + streamName
	}

	return streamName
}

func (j *jetStreamBroker) Publish(topic string, msg []byte, opts ...broker.PublishOption) error {
	j.RLock()
	defer j.RUnlock()

	if !j.connected {
		return errors.New("not connected")
	}

	streamName := createStreamName(topic)

	// 确保 stream 存在
	_, err := j.js.StreamInfo(streamName)
	if err != nil {
		// 如果 stream 不存在，创建它
		_, err = j.js.AddStream(&nats.StreamConfig{
			Name:     streamName,
			Subjects: []string{topic},
		})
		if err != nil {
			return err
		}
	}

	_, err = j.js.Publish(topic, msg)
	return err
}

func (j *jetStreamBroker) Subscribe(topic string, handler broker.Handler, opts ...broker.SubscribeOption) (broker.Subscriber, error) {
	j.RLock()
	defer j.RUnlock()

	if !j.connected {
		return nil, errors.New("not connected")
	}

	var opt broker.SubscribeOptions
	for _, o := range opts {
		o(&opt)
	}

	streamName := createStreamName(topic)

	// 确保 stream 存在
	_, err := j.js.StreamInfo(streamName)
	if err != nil {
		// 如果 stream 不存在，创建它
		_, err = j.js.AddStream(&nats.StreamConfig{
			Name:     streamName,
			Subjects: []string{topic},
		})
		if err != nil {
			return nil, err
		}
	}

	subOpts := []nats.SubOpt{
		nats.DeliverNew(),
		nats.AckExplicit(),
		nats.MaxDeliver(1),        // 设置为1，禁止重发
		nats.AckWait(time.Second), // 减少确认等待时间
	}

	if len(opt.Queue) > 0 {
		sub, err := j.js.QueueSubscribe(topic, opt.Queue, func(msg *nats.Msg) {
			pub := &jsPublication{t: msg.Subject, m: msg.Data, ctx: opt.Context, msg: msg}
			if err := handler(pub); err != nil {
				pub.err = err
				if j.opts.ErrorHandler != nil {
					j.opts.ErrorHandler(pub)
				}
			}
		}, subOpts...)
		if err != nil {
			return nil, err
		}
		return &jsSubscriber{sub: sub, opts: opt}, nil
	}

	sub, err := j.js.Subscribe(topic, func(msg *nats.Msg) {
		pub := &jsPublication{t: msg.Subject, m: msg.Data, ctx: opt.Context, msg: msg}
		if err := handler(pub); err != nil {
			pub.err = err
			if j.opts.ErrorHandler != nil {
				j.opts.ErrorHandler(pub)
			}
		}
	}, subOpts...)
	if err != nil {
		return nil, err
	}
	return &jsSubscriber{sub: sub, opts: opt}, nil
}

func (j *jetStreamBroker) setAddrs(addrs []string) []string {
	var cAddrs []string
	for _, addr := range addrs {
		if addr == "" {
			continue
		}
		if !strings.HasPrefix(addr, "nats://") {
			addr = "nats://" + addr
		}
		cAddrs = append(cAddrs, addr)
	}
	if len(cAddrs) == 0 {
		cAddrs = []string{nats.DefaultURL}
	}
	return cAddrs
}

func NewJetStreamBroker(opts ...broker.Option) broker.Broker {
	options := broker.Options{
		Context: context.Background(),
	}

	b := &jetStreamBroker{
		opts: options,
	}
	b.Init(opts...)
	return b
}
